#include <ConnectionBroker/PacketUtils.h>
#include <ConnectionBroker/TcpConnection.h>
#include <ConnectionBroker/server/TcpServer.h>
#include <boost/bind.hpp>
#include <boost/lexical_cast.hpp>
#include <json/json.h>

using namespace ConnectionBroker;
using boost::asio::ip::tcp;

TcpConnection::TcpConnection(bool clientMode,
                             boost::asio::io_service& pIoService, 
                             boost::shared_ptr<boost::asio::ssl::context> pContext)
: mClientMode(clientMode), mIoService(pIoService), mStopping(false), mUsingSsl(false)
{
   if(pContext)
   {
      mUsingSsl = true;
      mSocket = boost::shared_ptr<boost::asio::ip::tcp::socket>();
      mSslSocket.reset(new boost::asio::ssl::stream<boost::asio::ip::tcp::socket>(pIoService, *pContext));
      if(clientMode)
         mSslSocket->set_verify_mode(boost::asio::ssl::verify_none);
   }
   else
   {
      mSslSocket = boost::shared_ptr<boost::asio::ssl::stream<boost::asio::ip::tcp::socket> >();
      mSocket.reset(new boost::asio::ip::tcp::socket(pIoService));
   }
}

TcpConnection::~TcpConnection()
{
}

void TcpConnection::start()
{
   asyncHandshake();
}

void TcpConnection::connectionEstablished()
{
   asyncReadHeader();
}

void TcpConnection::handleHandshake(const boost::system::error_code& error)
{
   if(error)
   {
      std::cout << "handleHandshake error: " << error.message() << std::endl;
      onStop();
      return;
   }

   connectionEstablished();
}

void TcpConnection::handleReadBody(const boost::system::error_code& error)
{
   if(error)
   {
      std::cout << "handleReadBody error: " << error.message() << std::endl;
      onStop();
      return;
   }

   messageReceived(PacketUtils::tcpMessageToPacket(mReadMessage));
   asyncReadHeader();
}

void TcpConnection::asyncHandshake()
{
   if(!mSslSocket)
   {
      connectionEstablished();
      return;
   }

   boost::asio::ssl::stream_base::handshake_type handshakeType = boost::asio::ssl::stream_base::server;
   if(mClientMode)
      handshakeType = boost::asio::ssl::stream_base::client;

   mSslSocket->async_handshake(handshakeType,
        boost::bind(&TcpConnection::handleHandshake, shared_from_this(),
          boost::asio::placeholders::error));
}

void TcpConnection::asyncReadHeader()
{
   if(mSslSocket)
   {
      boost::asio::async_read(*mSslSocket,
      boost::asio::buffer(mReadMessage.header(), TcpMessage::header_length),
      boost::bind(&TcpConnection::handleReadHeader, shared_from_this(),
      boost::asio::placeholders::error));   
   }
   else
   {
      boost::asio::async_read(*mSocket,
      boost::asio::buffer(mReadMessage.header(), TcpMessage::header_length),
      boost::bind(&TcpConnection::handleReadHeader, shared_from_this(),
      boost::asio::placeholders::error));   
   }
}

void TcpConnection::asyncReadBody()
{
   if(mSslSocket)
   {
      boost::asio::async_read(*mSslSocket,
      boost::asio::buffer(mReadMessage.body(), mReadMessage.body_length()),
      boost::bind(&TcpConnection::handleReadBody, shared_from_this(),
      boost::asio::placeholders::error));   
   }
   else
   {
      boost::asio::async_read(*mSocket,
      boost::asio::buffer(mReadMessage.body(), mReadMessage.body_length()),
      boost::bind(&TcpConnection::handleReadBody, shared_from_this(),
      boost::asio::placeholders::error));   
   }
}

void TcpConnection::handleReadHeader(const boost::system::error_code& error)
{
   if(error)
   {
      std::cout << "handleReadHeader error: " << error.message() << std::endl;
      onStop();
      return;
   }

   mReadMessage.decode_header();
   asyncReadBody();
}

void TcpConnection::handleWrite(const boost::system::error_code& error)
{
   if(error)
   {
      std::cout << "handleWrite error: " << error.message() << std::endl;
      onStop();
      return;
   }

   mWriteMessages.pop_front();

   if(mWriteMessages.empty())
      return;

   asyncWrite(mWriteMessages.front());
}

void TcpConnection::handleConnect(const boost::system::error_code& error)
{
   if(error)
      return;

   asyncHandshake();
}

void TcpConnection::stop()
{
   if(mStopping)
      return;

   mIoService.post(boost::bind(&TcpConnection::onStop, shared_from_this()));
}

void TcpConnection::send(const Packet& pPacket)
{
   mIoService.post(boost::bind(&TcpConnection::doWrite, shared_from_this(), pPacket));
}

void TcpConnection::onStop()
{
   if(mSslSocket)
      mSslSocket->lowest_layer().close();
   else
      mSocket->close();
}

void TcpConnection::doWrite(Packet msg)
{
   bool pWriteInProgress = !mWriteMessages.empty();
   mWriteMessages.push_back(PacketUtils::packetToTcpMessage(msg));
   if(pWriteInProgress)
      return;

   asyncWrite(mWriteMessages.front());
}

void TcpConnection::asyncConnect(const std::string& address, const unsigned short port)
{
   tcp::resolver resolver(mIoService);
   std::string portStr(boost::lexical_cast<std::string>(port));
   tcp::resolver::query query(address, portStr);
   tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);

   if(mSslSocket)
   {
      boost::asio::async_connect(
         mSslSocket->lowest_layer(),
         endpoint_iterator,
         boost::bind(&TcpConnection::handleConnect, shared_from_this(), boost::asio::placeholders::error));
   }
   else
   {
      boost::asio::async_connect(
         *mSocket,
         endpoint_iterator,
         boost::bind(&TcpConnection::handleConnect, shared_from_this(), boost::asio::placeholders::error));
   }
}

void TcpConnection::asyncWrite(const TcpMessage& msgTcp)
{
   if(mSslSocket)
   {
      boost::asio::async_write(*mSslSocket,
         boost::asio::buffer(msgTcp.data(), msgTcp.length()),
         boost::bind(&TcpConnection::handleWrite, shared_from_this(),
         boost::asio::placeholders::error));
   }
   else
   {
      boost::asio::async_write(*mSocket,
         boost::asio::buffer(msgTcp.data(), msgTcp.length()),
         boost::bind(&TcpConnection::handleWrite, shared_from_this(),
         boost::asio::placeholders::error));
   }
}

